// $Id: reactor.c 33 2011-04-14 18:49:15Z cedric.shih@gmail.com $
/*
 * Copyright (C) 2010 Cedric Shih <cedric.shih@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include <string.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <event.h>

#include "nio.h"
#include "nio/socket.h"
#include "reactor.h"

#include "nio/test.h"

#define TIMER_COUNT 3
#define TIMER_PERIOD 1

#define PIPE_COUNT 5
#define CONCURRENCY 5

static struct event_base *evbase;
static struct nioreactor *reactor;
static struct nioradiator *radiator;
static struct niofuture *timeup;
static struct niosocket *in;
static struct niosocket *out;
static struct niofuture *readable;
static struct niofuture *writable;
static struct timeval timer;
static unsigned int counter;
static const char *data = "0123456789";
#define RECV 0
#define SEND 1
static int pipes[2];

//static int pipe_describe_read(void *channel)
//{
//	return ((int *)(channel))[RECV];
//}
//
//static ssize_t pipe_read(void *channel, void *buf, size_t count)
//{
//	return read(((int *)(channel))[RECV], buf, count);
//}
//
//static ssize_t pipe_write(void *channel, const void *buf, size_t count)
//{
//	return write(((int *)(channel))[SEND], buf, count);
//}
//
//static struct nio piperead = {
//	pipe_describe_read,
//	pipe_read,
//	NULL,
//};
//
//static int pipe_describe_write(void *channel)
//{
//	return ((int *)(channel))[SEND];
//}
//
//static struct nio pipewrite = {
//	pipe_describe_write,
//	NULL,
//	pipe_write,
//};
//
//ssize_t niosocket_write(void *channel, const void *buf, size_t count);

static void test_setup(void)
{
	evbase = event_base_new();
	radiator = nioradiator_new(CONCURRENCY);
	reactor = nioreactor_new(evbase);
	timeup = niofuture_new(reactor);
	in = niosocket_new();
	out = niosocket_new();
	readable = niofuture_new(reactor);
	writable = niofuture_new(reactor);
	counter = 0;
}

static void test_teardown(void)
{
	niofuture_free(writable);
	niofuture_free(readable);
	niosocket_free(out);
	niosocket_free(in);
	niofuture_free(timeup);
	nioreactor_free(reactor);
	nioradiator_free(radiator);
	event_base_free(evbase);
}

static void test_init(void)
{
	TEST_ASSERT_NOT_NULL(evbase);
	TEST_ASSERT_NOT_NULL(radiator);
	TEST_ASSERT_NOT_NULL(reactor);
}

static void test_timeup_cb(struct niofuture *future, struct niosocket *socket,
		int flags, void *arg)
{
	int rc;

	TEST_ASSERT(timeup == future);
	TEST_ASSERT_NULL(socket);
	TEST_ASSERT_EQUAL_INT(NIO_TIMEUP, flags);
	TEST_ASSERT_NULL(arg);

	counter++;

	if (counter < TIMER_COUNT) {
		rc = niofuture_timeup(future, &timer, test_timeup_cb, NULL);
		TEST_ASSERT_EQUAL_INT(0, rc);
	} else {
		rc = nioreactor_break(reactor);
		TEST_ASSERT_EQUAL_INT(0, rc);
	}
}

static void test_timeup_do(int concurrent)
{
	int rc;
	struct timeval begin;
	struct timeval end;

	rc = gettimeofday(&begin, NULL);
	TEST_ASSERT_EQUAL_INT(0, rc);

	timer.tv_sec = TIMER_PERIOD;
	timer.tv_usec = 0;
	counter = 0;

	rc = niofuture_timeup(timeup, &timer, test_timeup_cb, NULL);
	TEST_ASSERT_EQUAL_INT(0, rc);

	if (concurrent) {
		rc = nioradiator_start(radiator);
		TEST_ASSERT_EQUAL_INT(0, rc);

		rc = nioreactor_loop(reactor, radiator);
		TEST_ASSERT_EQUAL_INT(0, rc);

		rc = nioradiator_stop(radiator);
		TEST_ASSERT_EQUAL_INT(0, rc);
	} else {
		rc = nioreactor_loop(reactor, NULL);
		TEST_ASSERT_EQUAL_INT(0, rc);
	}

	rc = gettimeofday(&end, NULL);
	TEST_ASSERT_EQUAL_INT(0, rc);

	TEST_ASSERT((end.tv_sec - begin.tv_sec) >=
			(TIMER_PERIOD * TIMER_COUNT));
}

static void test_timeup(void)
{
	test_timeup_do(0);
}

static void test_radiator_timeup(void)
{
	test_timeup_do(1);
}

static void test_pipe_read_cb(struct niofuture *future,
		struct niosocket *socket, int flags, void *arg);

static void test_pipe_write_cb(struct niofuture *future,
		struct niosocket *socket, int flags, void *arg)
{
	int rc;
	ssize_t len;

	TEST_ASSERT(writable == future);
	TEST_ASSERT(socket == out);
	TEST_ASSERT_EQUAL_INT(NIO_WRITABLE, flags);
	TEST_ASSERT_NULL(arg);

	len = niosocket_write(socket, data, strlen(data));
	TEST_ASSERT_EQUAL_INT(strlen(data), len);

	rc = niofuture_readable(readable, in, NULL,
			test_pipe_read_cb, NULL);
	TEST_ASSERT_EQUAL_INT(0, rc);
}

static void test_pipe_read_cb(struct niofuture *future,
		struct niosocket *socket, int flags, void *arg)
{
	int rc;
	ssize_t len;
	char buffer[256];
	const char *actual = buffer;

	TEST_ASSERT(readable == future);
	TEST_ASSERT(socket == in);
	TEST_ASSERT_EQUAL_INT(NIO_READABLE, flags);
	TEST_ASSERT_NULL(arg);

	len = niosocket_read(socket, buffer, sizeof(buffer));
	TEST_ASSERT_EQUAL_INT(strlen(data), len);
	buffer[len] = '\0';
	TEST_ASSERT_EQUAL_STRING(data, actual);

	counter++;

	if (counter < PIPE_COUNT) {
		rc = niofuture_writable(writable, out, NULL,
				test_pipe_write_cb, NULL);
		TEST_ASSERT_EQUAL_INT(0, rc);
	} else {
		rc = nioreactor_break(reactor);
		TEST_ASSERT_EQUAL_INT(0, rc);
	}
}

static void test_pipe_do(int concurrent)
{
	int rc;

	rc = pipe(pipes);
	TEST_ASSERT_EQUAL_INT(0, rc);

	rc = niosocket_attach(in, pipes[RECV]);
	TEST_ASSERT_EQUAL_INT(0, rc);

	rc = niosocket_attach(out, pipes[SEND]);
	TEST_ASSERT_EQUAL_INT(0, rc);

	rc = niofuture_writable(writable, out, NULL,
			test_pipe_write_cb, NULL);
	TEST_ASSERT_EQUAL_INT(0, rc);

	if (concurrent) {
		rc = nioradiator_start(radiator);
		TEST_ASSERT_EQUAL_INT(0, rc);

		rc = nioreactor_loop(reactor, radiator);
		TEST_ASSERT_EQUAL_INT(0, rc);

		rc = nioradiator_stop(radiator);
		TEST_ASSERT_EQUAL_INT(0, rc);
	} else {
		rc = nioreactor_loop(reactor, NULL);
		TEST_ASSERT_EQUAL_INT(0, rc);
	}
}

static void test_pipe(void)
{
	test_pipe_do(0);
}

static void test_radiator_pipe(void)
{
	test_pipe_do(1);
}

TestRef nioreactor_test(void)
{
	EMB_UNIT_TESTFIXTURES(fixtures) {
		new_TestFixture("nioreactor_test_init", test_init),
		new_TestFixture("nioreactor_test_timeup", test_timeup),
		new_TestFixture("nioreactor_test_radiator_timeup",
				test_radiator_timeup),
		new_TestFixture("nioreactor_test_pipe", test_pipe),
		new_TestFixture("nioreactor_test_radiator_pipe",
				test_radiator_pipe),
	};

	EMB_UNIT_TESTCALLER(test_case, "nioreactor_test_case", test_setup,
			test_teardown, fixtures);

	return (TestRef)&test_case;
}
